#pragma once

#include <functional>
#include <cstdio>
#include <atomic>
#include <thread>
#include <mutex>
#include <memory>
#include <string>
#include <sys/epoll.h>

#include <sys/types.h>
#include <sys/socket.h>

#include "log/log.h"
#include "nanomsg/nn.h"
#include "nanomsg/pair.h"
#include "nanomsg/pubsub.h"
#include "nanomsg/pipeline.h"
#include "nanomsg/tcp.h"

class SmartNetPubsub
{
public:
    enum FS {
        NP_NONE,
        NP_PUB,
        NP_SUB
    };

public:
    SmartNetPubsub(enum FS fs, std::string sockUrl) : m_fs(fs), m_sockUrl(sockUrl) {
        init();
    }
    ~SmartNetPubsub() {
        deinit();
    }

    int init() {
        if (NP_PUB == m_fs) {
            return init_pub();
        }
        else if (NP_SUB == m_fs) {
            return init_sub(16*1024*1024);
        }
        else {
            log_error("network_pubsub init failed.\n");
            exit(EXIT_FAILURE);
        }
        
        return -1;
    }

    int deinit() {
        if (m_sockFd >= 0) {
            nn_shutdown(m_sockFd, 0);
        }
        return 0;
    }

    int stop(void) {
        m_loop_run = false;
        std::lock_guard<std::mutex> locker(m_mutex);
        return 0;
    }

    int send_mesg(uint8_t *data, uint32_t len) {
        if (NP_PUB == m_fs) {
            return pub_msg(data, len);
        }
        else {
            log_error("not PUB mode, can't use SendMesg.");
            exit(EXIT_FAILURE);
        }
        return -1;
    }

    int loop_recv_mesg(std::function<void(uint8_t *, uint32_t)> callback) {
        if (NP_SUB == m_fs) {
            return loop_sub_recv_msg(callback);
        }
        else {
            log_error("not SUB mode, can't use LoopRecvMesg.");
            exit(EXIT_FAILURE);
        }
        return -1;
    }

private:
    int init_pub() {
        int ret;

        m_sockFd = nn_socket(AF_SP, NN_PUB);
        if (m_sockFd < 0) {
            log_error("nn_socket: %s\n", nn_strerror (nn_errno ()));
            exit(EXIT_FAILURE);
        }
        log_debug_q(QUIET_LOG_DEBUG, "nn_socket seccess, sockFd:%d.", m_sockFd);

        ret = nn_bind(m_sockFd, m_sockUrl.c_str());
        if (ret < 0) {
            log_error("nn_bind: %s\n", nn_strerror (nn_errno ()));
            nn_close (m_sockFd);
            exit(EXIT_FAILURE);
        }
        log_debug_q(QUIET_LOG_DEBUG, "nn_bind seccess, URL:%s.", m_sockUrl.c_str());

        return 0;
    }
    
    int pub_msg(uint8_t *data, uint32_t len) {
        int ret;
    #if 1
        ret = nn_send(m_sockFd, data, len, 0);
        if (ret < 0) {
            log_error("nn_send: %s (ignoring)\n", nn_strerror (nn_errno ()));
        }
        log_debug_q(QUIET_LOG_DEBUG, "nn_send secess.");
    #else  // 分段发送
        uint8_t offset = 0;
        do {
            if ((len - offset) >= 1024*1024) {
                ret = nn_send(m_sockFd, data + offset, 1024*1024, 0);
            }
            else {
                ret = nn_send(m_sockFd, data + offset, len - offset, 0);
            }

            if (ret < 0) {
                log_error("nn_send: %s (ignoring)\n", nn_strerror (nn_errno ()));
            }

            offset += ret;
        } while (offset < len);
        return offset;

    #endif
        return ret;
    }

    int init_sub(uint32_t buff_max_size) {
        int ret;

        m_sockFd = nn_socket(AF_SP, NN_SUB);
        if (m_sockFd < 0) {
            log_error("nn_socket: %s\n", nn_strerror (nn_errno ()));
            exit(EXIT_FAILURE);
        }
        log_debug_q(QUIET_LOG_DEBUG, "nn_socket seccess, sockFd:%d.", m_sockFd);

        // 网络缓存区大小设置
        ret = nn_setsockopt (m_sockFd, NN_SOL_SOCKET, NN_RCVMAXSIZE, &buff_max_size, sizeof(buff_max_size));
        if (ret < 0) {
            log_error("nn_setsockopt: %s\n", nn_strerror (nn_errno ()));
            nn_close (m_sockFd);
            exit(EXIT_FAILURE);        
        }
        log_debug_q(QUIET_LOG_DEBUG, "set NN_RCVMAXSIZE seccess, buff_max_size:%d.", buff_max_size);

        ret = nn_connect(m_sockFd, m_sockUrl.c_str());
        if (ret < 0) {
            log_error("nn_socket: %s\n", nn_strerror (nn_errno ()));
            nn_close (m_sockFd);
            exit(EXIT_FAILURE);
        }
        log_debug_q(QUIET_LOG_DEBUG, "nn_connect seccess, URL:%s.", m_sockUrl.c_str());

        ret = nn_setsockopt(m_sockFd, NN_SUB, NN_SUB_SUBSCRIBE, "", 0);
        if (ret < 0) {
            log_error("nn_setsockopt: %s\n", nn_strerror (nn_errno ()));
            nn_close (m_sockFd);
            exit(EXIT_FAILURE);
        }

        return 0;
    }

    int loop_sub_recv_msg(std::function<void(uint8_t *, uint32_t)> callback) {
        std::lock_guard<std::mutex> locker(m_mutex);
        int buf_len;
        int ret;
        uint8_t *buf;

        nn_pollfd pfd;
        pfd.fd = m_sockFd;
        pfd.events = NN_POLLIN;
        int timeout = 1000;  // ms

        if (-1 >= m_sockFd) {
            log_error("sock fd not init.\n");
            exit(EXIT_FAILURE);
        }

        while (true == m_loop_run) {
            ret = nn_poll(&pfd, 1, timeout);
            if (ret <= 0) {
                log_debug_q(QUIET_LOG_DEBUG, "nn_poll timeout.");
                continue;
            }

            if (pfd.revents & NN_POLLIN) {
                buf_len = nn_recv(m_sockFd, &buf, NN_MSG, NN_DONTWAIT);
                if (buf_len > 0) {
                    callback(buf, buf_len);
                    nn_freemsg(buf);
                }
                else {
                    log_debug_q(QUIET_LOG_DEBUG, "nn_recv nullptr ret:%d.", buf_len);
                }
            }
        }

        return 0;
    }

private:
    enum FS m_fs{NP_NONE};
    int m_sockFd{-1};    
    std::string m_sockUrl;
    std::atomic_bool m_loop_run{true};
    std::mutex m_mutex;

    /* logger config */
    const int QUIET_LOG_DEBUG{1};
    const int QUIET_LOG_INFO{1};
};